package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.common.GmallConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.flink.util.StringUtils;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.List;

/**
 * @author: xu
 * @desc: 动态分流实现
 */
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
    /**
     * 声明维度侧输出流标签
     */
    private OutputTag<JSONObject> hbaseTag;
    /**
     * 声明广播状态描述器
     */
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    /**
     * 声明连接对象
     */
    private Connection conn = null;

    public TableProcessFunction(OutputTag<JSONObject> hbaseTag, MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.hbaseTag = hbaseTag;
        this.mapStateDescriptor = mapStateDescriptor;
    }

    /**
     * 在函数被调用的时候执行的方法，执行一次
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        // 初始化Phoenix连接
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    /**
     * 处理业务流中数据，每来一个元素执行一次，主要是根据配置表Map对当前进来的元素进行分流处理
     */
    @Override
    public void processElement(JSONObject jsonObj, BroadcastProcessFunction<JSONObject, String, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        // 获取表名，操作类型
        String table = jsonObj.getString("table");
        String type = jsonObj.getString("type");

        // 注意：问题修复，如果使用Maxwell的Bootstrap同步历史数据，这个时候它的操作类型叫bootstrap-insert
        if ("bootstrap-insert".equals(type)) {
            type = "insert";
            jsonObj.put("type", type);
        }
        // 根据表名和操作类型拼接key
        String key = table + ":" + type;
        // 获取状态，并根据Key从状态中获取配置信息
        TableProcess tableProcess = ctx.getBroadcastState(mapStateDescriptor).get(key);

        // TODO：如果获取到了该元素对应的配置信息
        if (tableProcess != null) {
            String sinkColumns = tableProcess.getSinkColumns();
            // 如果指定了sinkColumn，需要对保留的字段进行过滤处理
            if (!StringUtils.isNullOrWhitespaceOnly(sinkColumns)) {
                filterColumn(jsonObj.getJSONObject("data"), sinkColumns);
            }
            // TODO：获取sinkTable，指明当前数据发往哪里，维度数据就是phoenix中的表名，事实数据就是kafka的主题名
            jsonObj.put("sink_table", tableProcess.getSinkTable());
            String sinkType = tableProcess.getSinkType();
            if (TableProcess.SINK_TYPE_HBASE.equals(sinkType)) {
                // 如果sinkType = hbase，说明是维度数据，通过侧输出流输出
                ctx.output(hbaseTag, jsonObj);
            } else if (TableProcess.SINK_TYPE_KAFKA.equals(sinkType)) {
                // 如果sinkType = kafka，说明是事实数据，通过主流输出
                out.collect(jsonObj);
            }
        } else {
            // 在配置表中没有该操作对应的配置
            System.out.println("No this Key in TableProcess:" + key);
        }
    }

    /**
     * 对Data中数据进行进行过滤
     */
    private void filterColumn(JSONObject data, String sinkColumns) {
        // sinkColumns表示要保留那些列，id,out_trade_no,order_id
        String[] cols = sinkColumns.split(",");
        // 将数组转换为集合，为了判断集合中是否包含某个元素
        List<String> columnList = Arrays.asList(cols);
        // 获取json对象中封装的一个个键值对，每个键值对封装为Entry类型
        data.entrySet().removeIf(entry -> !columnList.contains(entry.getKey()));
    }

    /**
     * 处理广播流中数据，FlinkCDC从MySQL中读取配置信息
     */
    @Override
    public void processBroadcastElement(String jsonStr, BroadcastProcessFunction<JSONObject, String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
        // 将json格式字符串转换为JSON对象
        JSONObject jsonObj = JSON.parseObject(jsonStr);
        // 获取配置表中一条配置信息，封装成TableProcess
        TableProcess tableProcess = JSON.parseObject(jsonObj.getString("data"), TableProcess.class);

        // TODO：获取TableProcess中的7类属性
        String sourceTable = tableProcess.getSourceTable();
        String operateType = tableProcess.getOperateType();
        String sinkType = tableProcess.getSinkType();
        String sinkTable = tableProcess.getSinkTable();
        String sinkColumns = tableProcess.getSinkColumns();
        String sinkPk = tableProcess.getSinkPk();
        String sinkExtend = tableProcess.getSinkExtend();

        // 如果读取到的配置是发送到HBase的话，要先将维度表创建出来
        if (sinkType.equals(TableProcess.SINK_TYPE_HBASE) && "insert".equals(operateType)) {
            // 检查Phoenix中是否存在这种表，不存在那么需要将表创建出来
            checkTable(sinkTable, sinkColumns, sinkPk, sinkExtend);
        }

        // 拼接保存配置的key
        String key = sourceTable + ":" + operateType;
        // 获取状态，并将配置信息放到状态中
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        broadcastState.put(key, tableProcess);
    }

    /**
     * 检查表是否存在，不存在使用Phoenix建立维度表：create table if not exists 表空间.表名(字段名 数据类型,字段名 数据类型);
     */
    private void checkTable(String tableName, String fields, String pk, String ext) throws SQLException {
        // 如果在配置表中，没有配置主键需要给一个默认主键的值
        if (pk == null) pk = "id";
        // 如果在配置表中，没有配置建表扩展需要给一个默认建表扩展的值
        if (ext == null) ext = "";
        // 拼接建表语句
        StringBuilder createTableSql = new StringBuilder("create table if not exists ")
                .append(GmallConfig.HBASE_SCHEMA).append(".")
                .append(tableName).append("(");
        // 对建表字段进行切分，然后遍历
        String[] fieldsArr = fields.split(",");
        for (int i = 0; i < fieldsArr.length; i++) {
            String field = fieldsArr[i];
            // 判断当前字段是否为主键字段
            if (pk.equals(field)) {
                createTableSql.append(field).append(" varchar primary key ");
            } else {
                createTableSql.append(field).append(" varchar ");
            }
            if (i < fieldsArr.length - 1) {
                createTableSql.append(",");
            }
        }
        createTableSql.append(")").append(ext);
        System.out.println("Phoenix中的建表语句:" + createTableSql);
        // 获取Phoenix连接
        PreparedStatement ps = null;
        try {
            ps = conn.prepareStatement(createTableSql.toString());
            ps.execute();
        } catch (SQLException e) {
            e.printStackTrace();
            throw new RuntimeException("在Phoenix中建表失败");
        } finally {
            if (ps != null && !ps.isClosed()) {
                ps.close();
            }
        }
    }
}
